package com.moli.iov.handle.real.mq.kafka;

import com.moli.iov.cache.RedisUtil;
import com.moli.iov.consumer.KafkaConsumerServer;
import com.moli.iov.contant.RealContant;
import com.moli.iov.handle.real.util.Contact;
import com.moli.iov.protocol.codec.DataDecoder;
import com.moli.iov.protocol.codec.ReqStreamData;
import com.moli.iov.protocol.enums.CommandTag;
import com.moli.iov.protocol.enums.realtime.LocationStateTag;
import com.moli.iov.protocol.model.realtime.*;
import com.moli.iov.util.ByteArrayUtil;
import com.moli.iov.util.GsonUtil;
import lombok.Data;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.support.Acknowledgment;

import javax.annotation.Resource;
import java.util.Date;
import java.util.HashMap;
import java.util.Map;

@Data
public class RealTimeConsumer extends KafkaConsumerServer {

    private static final String KEY_LOCATIONSTATE = "IOV:PROTOCOL:LOCATIONSTATE";

    private static final String SPLIT = "_";

    @Resource
    RedisUtil<String, String> redisUtil;

    @Override
    public void manualHandle(ConsumerRecord<String, Object> record,
                             Acknowledgment acknowledgment) {

        long start = System.currentTimeMillis();
        try {
            ReqStreamData data = (ReqStreamData) record.value();
            if (data != null) {
                RealTimeInfoReq realTimeReq = new RealTimeInfoReq();
                //解析ReqStreamData中的报文
                DataDecoder<RealTimeInfoReq> decoder = new DataDecoder<RealTimeInfoReq>();
                decoder.decode(realTimeReq, data.getDataBytes());

                if (realTimeReq != null && realTimeReq.commandTag != null && realTimeReq.commandTag.equals(CommandTag.REALTIME_INFO_REPORT)) {
                    String realTimeJson = buildJson(record.key(), realTimeReq, data.getCollectDate());
                    redisUtil.hset(RealContant.KEY_MSG_REALTIME, record.key(), realTimeJson);
                    String mess = ByteArrayUtil.toHexString(data.getDataBytes());
                    redisUtil.hset(RealContant.KEY_ORIGINAL_REALTIME, record.key(), mess);

                    if (realTimeReq.getVehicleGpsReq() != null
                            && realTimeReq.getVehicleGpsReq().getLocationState() != null) {
                        LocationStateTag.IsValid isValid = realTimeReq.getVehicleGpsReq().getLocationState().getIsValid();
                        if (isValid != null) {
                            redisUtil.hset(KEY_LOCATIONSTATE, record.key(), record.key()+SPLIT+isValid.getValue());
                        }
                    }

                    logger.info("[车辆实时原始报文][vin:{}][mess:{}][offset:{}]", record.key(), mess, record.offset());
                }
            } else {
                logger.error("[车辆实时报文][数据为空]{}", record.key(), record.value());
            }
        } catch (
                Exception e)

        {
            logger.error("[车辆实时报文 异常][vin={}][data={}][offset={}][err={}]", record.key(), record.value(), record.offset(), e.getMessage());
        }

        long time = System.currentTimeMillis() - start;
        logger.info("[车辆:{}，实时报文:{}][时长:{}ms]", record.key(), record.value(), time);
        acknowledgment.acknowledge();
    }

    public static void main(String[] args) {
        byte b = 0x01;
        System.out.println(b);

    }
    /**
     * 将报文对象转换为业务json
     */
    public String buildJson(String vin, RealTimeInfoReq realTimeInfoReq, Date collectDate) {

        if (collectDate != null && realTimeInfoReq != null) {
            Map<String, Object> map = new HashMap<String, Object>();

            map.put(Contact.VIN, vin);
            //采集时间
            map.put(Contact.COLLECT_DATE, collectDate);

            map.put(Contact.REPORT_DATE, realTimeInfoReq.collectDate == null ? null : realTimeInfoReq.collectDate.getDate());

            //GPS数据
            VehicleGpsReq gpsReq = realTimeInfoReq.getVehicleGpsReq();

            if (gpsReq != null) {
                map.put(Contact.VEHICLE_GPS, gpsReq);
            }
            //整车数据
            VehicleAllReq allReq = realTimeInfoReq.getVehicleAllReq();
            if (allReq != null) {
                map.put(Contact.VEHICLE_ALL, allReq);
                //				BeanUtils.copyProperties(gpsReq, real.getVehicleAll());
            }

            //扩展数据
            ExtendReq extendReq = realTimeInfoReq.getExtendReq();
            if (extendReq != null) {
                //扩展车辆状态数据
                ExtendVehicleState extendVehicleState = extendReq.getextendVehicleState();
                if (extendVehicleState != null) {
                    //扩展车辆状态数据
                    map.put(Contact.EXTEND_VEHICLE_STATE, extendVehicleState);
                    //					result.append("\"extendVehicleState\":");
                    //					BeanUtils.copyProperties(extendVehicleState, real.getExtendVehicleState());
                }
            }
            ;


            //驱动电机数据
            DriveMotorReq motorReq = realTimeInfoReq.getDriveMotorReq();
            if (motorReq != null) {
                //驱动电机数据
                map.put(Contact.DRIVE_MOTOR, motorReq);
            }

            //			//驱动电机数据
            //			DriveMotorReq motorReq = realTimeInfoReq.getDriveMotorReq();
            //			if(motorReq!=null){
            //				List<DriveMotorInfo> list=motorReq.getListDriveMotors();
            //				List<DriveMotorInfoDto>  driveMotorInfoDtoList = new ArrayList<DriveMotorInfoDto>();
            //				if(list!=null && list.size()>0){
            //					for (DriveMotorInfo driveMotorInfo : list) {
            //						DriveMotorInfoDto driveMotorInfoDto = new DriveMotorInfoDto();
            //						BeanUtils.copyProperties(driveMotorInfo, driveMotorInfoDto);
            //						driveMotorInfoDtoList.add(driveMotorInfoDto );
            //					}
            //				}
            //				realTimeInfoDto.getDriveMotor().setDriveMotorSum(motorReq.getDriveMotorSum());
            //			}
            return GsonUtil.toJsonWithSerializeNulls(map);
        }
        return null;
    }

}
